package comparator;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.StringTokenizer;

import matchers.JaroWinkler;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import file.FileArray;
import file.FileSD;

public class BDM {

	public static IntWritable partitionIndex;
	public static long sumPairs;
	public static ArrayList<BDMData> BDMData = new ArrayList<BDMData>();
	public static int blockingPos = 0;
	public static HashMap<String, Integer> blockingkeyIndexer = new HashMap<String, Integer>();
	
	public static class Map extends Mapper<LongWritable, Text, Text, FileSD> {

		private final static IntWritable one = new IntWritable(1);
		private int numberOfReplications = 5;
		private int token = 5;
		private Text word = new Text();
		private Text k = new Text();

		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String pathName = ((FileSplit) context.getInputSplit()).getPath().getName();
			int numberOfFile = getNumberOfFile(pathName);

			StringTokenizer itr = new StringTokenizer(value.toString());
			while (itr.hasMoreTokens()) {
				word.set(itr.nextToken());
				if (numberOfFile == 2) {
					for (int i = 0; i < numberOfReplications; i++) {
						k.set(i +""/*+ "." + numberOfFile*/);
						context.write(k, new FileSD(word, new IntWritable(numberOfFile)));
					}
				} else {
					k.set(--token +""/*+ "." + numberOfFile*/);
					context.write(k, new FileSD(word, new IntWritable(numberOfFile)));
				}
				
			}
		}

		private int getNumberOfFile(String pathName) {
			return Character.getNumericValue(pathName.charAt(3));
		}


	}

	public static class Reduce extends
			Reducer<Text, FileSD, Text, Text> {
		private Text result = new Text();
		private Text k = new Text();
		private int log = 1;

		@Override
		public void reduce(Text key, Iterable<FileSD> values,
				Context context) throws IOException, InterruptedException {
			
			FileArray filesDataSource =  null;
			List<FileArray> filesDataSetTarget =  new ArrayList<FileArray>();
			for (FileSD val : values) {
				if (val.getDataset().get() == 1) {
					filesDataSource = new FileArray(val);
				} else {
					filesDataSetTarget.add(new FileArray(val));
				}
			}
			
			String saida = "";
			if (filesDataSource != null) {
				for (FileArray fileTarget : filesDataSetTarget) {
					saida += filesDataSource.getName() + " - " + fileTarget.getName() + " = " + ComputeMatchers(filesDataSource.getName(), fileTarget.getName()) + "\n";
				}
			}
			result.set(saida);
			k.set(log++ + "");
			context.write(k, result);	
	
		}



		private String ComputeMatchers(String filesDataSource, String fileTarget) {
			JaroWinkler winkler = new JaroWinkler();
			double result = winkler.compare(filesDataSource.toString(), fileTarget.toString());
			return new DecimalFormat("#.##").format(result);
		}

	}
	
}